package net.bwie.flink;

import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * Flink CEP 案例：30秒内连续3次登录失败，给出警告，当天不允许登录
 * @author xuanyu
 * @date 2025/10/18
 */
public class _03FlinkCepDemo {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3. 数据转换-transformation
		/*
			todo 检测用户行为，如果连续三次登录失败，并且30秒内，就输出报警信息。
user_1,192.168.0.1,fail,2025-10-18 11:35:12
user_1,192.168.0.1,fail,2025-10-18 11:35:24
user_1,192.168.0.1,fail,2025-10-18 11:35:40
user_2,192.168.0.1,success,2025-10-18 11:35:41
		 */
		// 3-1. 封装实体类
		SingleOutputStreamOperator<EventBean> stream31 = stream.map(
			new RichMapFunction<String, EventBean>() {
				@Override
				public EventBean map(String value) throws Exception {
					// 分割字符串
					String[] split = value.split(",");
					// 实体类对象
					EventBean bean = new EventBean(split[0], split[1], split[2], split[3]);
					// 返回对象
					return bean;
				}
			}
		);

		/*
			Flink CEP 开发编程5步法：
		        step1. 指定数据事件时间
		        step2. 分组KeyBy
		        step3. 指定规则
		        step4. 匹配规则
		        step5. 处理数据
		 */
		// todo step1. 指定数据事件时间
		SingleOutputStreamOperator<EventBean> stream_s1 = stream31.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<EventBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<EventBean>() {
					SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
					@SneakyThrows
					@Override
					public long extractTimestamp(EventBean element, long recordTimestamp) {
						// 获取登录时间
						String loginTime = element.getLoginTime();
						// 转换日期
						Date date = format.parse(loginTime);
						// long类型
						return date.getTime();
					}
				})
		);

		// todo step2. 分组KeyBy：按照用户分组
		KeyedStream<EventBean, String> stream_s2 = stream_s1.keyBy(
			new KeySelector<EventBean, String>() {
				@Override
				public String getKey(EventBean value) throws Exception {
					return value.getUserName();
				}
			}
		);

		// todo step3. 指定规则
		Pattern<EventBean, EventBean> pattern = Pattern
			// 规则：第1个条件
			.<EventBean>begin("login").where(
				new SimpleCondition<EventBean>() {
					@Override
					public boolean filter(EventBean value) throws Exception {
						// 登录状态
						String loginStatus = value.getLoginStatus();
						// 比较
						return "fail".equals(loginStatus);
					}
				}
			)
			// 规则：次数
			.times(3)
			// 规则：时间范围内
			.within(Time.seconds(30));


		// todo step4. 匹配规则
		PatternStream<EventBean> stream_s4 = CEP.pattern(stream_s2, pattern);

		// todo step5. 处理数据
		SingleOutputStreamOperator<String> stream_s5 = stream_s4.process(
			new PatternProcessFunction<EventBean, String>() {
				@Override
				public void processMatch(Map<String, List<EventBean>> match,
				                         Context ctx,
				                         Collector<String> out) throws Exception {
					/*
						Map<String, List<EventBean>> match 表示匹配规则数据
							map中Key：String字符串类型，表示规则名称
							map中value：List列表类型，存储多条符合规则的数据
					 */
					// 依据规则名称获取匹配数据
					List<EventBean> list = match.get("login");
					// 打印数据
					for (EventBean bean : list) {
						System.out.println("登录日志：" + bean);
					}
					// 输出
					out.collect("用户在30秒内，连续登录3次失败。。。。。。。。。。。。。");
				}
			}
		);

		// 4. 数据输出-sink
		stream_s5.print();

		// 5. 触发执行-execute
		env.execute("FlinkCepDemo") ;
	}

}
